package com.kyrie.kafka.kafkaConfig;

import com.kyrie.kafka.KafKaConfiguration;
import com.kyrie.kafka.KafkaProducerListener;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.*;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;

import java.util.HashMap;
import java.util.Map;

/***
 * 描述: kafka配置类
 *
 * @author wuxiang
 * @date 2020-04-24 13:43
 */
@Configuration
@ConditionalOnProperty(value = {"interface.kafka.enable"}, matchIfMissing = true)
public class KafkaConfig {

    @Autowired
    private KafKaConfiguration kafKaConfiguration;

    /**######################################## KAFKA_PRODUCER_CONFIG  START ########################################*/
    /**
     * 生产者配置
     */
    public Map<String, Object> producerConfigs() {

        Map<String, Object> props = new HashMap<>();
        // 集群的服务器地址
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafKaConfiguration.getBootstrapServers());
        //  消息缓存
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, kafKaConfiguration.getProducer().getBufferMemory());
        // 生产者空间不足时，send()被阻塞的时间，默认60s
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, kafKaConfiguration.getProducer().getMaxBlock());
        // 生产消息发送成功回调方案
        props.put(ProducerConfig.ACKS_CONFIG,kafKaConfiguration.getProducer().getAcks());
        // 失败重试次数
        props.put(ProducerConfig.RETRIES_CONFIG, kafKaConfiguration.getProducer().getRetries());
        // 失败重试的时间间隔
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG,kafKaConfiguration.getProducer().getRetriesBackoffMs());

        // 指定ProducerBatch（消息累加器中BufferPool中的）可复用大小
        props.put(ProducerConfig.BATCH_SIZE_CONFIG,  kafKaConfiguration.getProducer().getBatchSize());
        // 生产者会在ProducerBatch被填满或者等待超过LINGER_MS_CONFIG时发送
        props.put(ProducerConfig.LINGER_MS_CONFIG, kafKaConfiguration.getProducer().getLinger());
        // key 和 value 的序列化
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // 客户端id
        props.put(ProducerConfig.CLIENT_ID_CONFIG, kafKaConfiguration.getProducer().getClientId());

        return props;
    }

    /**
     * 生产者工厂
     */
    @Bean
    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    /**
     * kafkaTemplate 设置生产者监听
     */
    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(producerFactory());
        kafkaTemplate.setProducerListener(new KafkaProducerListener());
        return kafkaTemplate;
    }

    /**######################################## KAFKA_PRODUCER_CONFIG  END ########################################*/


    /**######################################## KAFKA_CONSUMER_CONFIG  START ########################################*/

    /**
     * 消费者配置：批量消费自动提交 或 批量消费手动提交
     */
    public Map<String,Object> consumerConfigs() {
        Map<String, Object> configProps = new HashMap<>();
        configProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafKaConfiguration.getBootstrapServers());
        configProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configProps.put(ConsumerConfig.GROUP_ID_CONFIG, kafKaConfiguration.getConsumer().getGroupId());
        // 批量消费消息数量
        configProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafKaConfiguration.getConsumer().getMaxPollRecords());
        // 自动提交偏移量
        // 如果设置成true,偏移量由auto.commit.interval.ms控制自动提交的频率
        // 如果设置成false,不需要定时的提交offset，可以自己控制offset，当消息认为已消费过了，这个时候再去提交它们的偏移量。
        // 这个很有用的，当消费的消息结合了一些处理逻辑，这个消息就不应该认为是已经消费的，直到它完成了整个处理。
        configProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, kafKaConfiguration.getConsumer().getEnableAutoCommit());
        // 自动提交的频率
        configProps.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, kafKaConfiguration.getConsumer().getAutoCommitInterval());
        // Session超时设置
        configProps.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, kafKaConfiguration.getConsumer().getSessionTimeout());
        // 该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理：
        // latest（默认值）在偏移量无效的情况下，消费者将从最新的记录开始读取数据（在消费者启动之后生成的记录）
        // earliest ：在偏移量无效的情况下，消费者将从起始位置读取分区的记录
        configProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, kafKaConfiguration.getConsumer().getAutoOffsetReset());

        return configProps;
    }


    /**
     * 消费者工厂
     */
    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    /**
     * 消费者监听器
     */
    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {

        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        // 设置消费者工厂
        factory.setConsumerFactory(consumerFactory());
        // 要创建的消费者数量(10 个线程并发处理)
        factory.setConcurrency(10);

        // 设置poll超时时间
        factory.getContainerProperties().setPollTimeout(1500);

        // 设置为批量消费，每个批次数量在Kafka配置参数中设置（max.poll.records）
        factory.setBatchListener(kafKaConfiguration.getConsumer().getEnableBatch());
        // 下句在手动提交位移时生效
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);

        return factory;
    }

    /**######################################## KAFKA_CONSUMER_CONFIG  END ########################################*/
}
